package org.example.flinktest.apidemo;

import java.util.ArrayList;
import java.util.List;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.example.flinktest.bean.SensorReading;
import org.example.flinktest.operators.MakeSensorMapFunction;

/**
 * @author shihb
 * @date 2019/12/10 11:02
 */
public class TransformDemo {

  public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    DataStream<String> streamFromFile = env.readTextFile(
        "D:\\SHBData\\IDEAProjects\\fink-parent\\flink-quickstart-java\\src\\main\\resources\\sensor.txt");

    //1.基本转换算子和简单聚合算子
    DataStream<SensorReading> sensorReadingStream = streamFromFile.map(new MakeSensorMapFunction());

    DataStream<SensorReading> aggStream = sensorReadingStream.keyBy("id")
//        .sum("temperature")
        .reduce(new ReduceFunction<SensorReading>() {
          @Override
          public SensorReading reduce(SensorReading sr1, SensorReading sr2)
              throws Exception {
            return new SensorReading(sr1.getId(), sr2.getTimestamp() + 1,
                (sr1.getTemperature() + sr2.getTemperature()) / 2);
          }
        });
//    aggStream.print();

    //2.多流转换算子
    /**
     * split分流后变成splitStream要进行select后才会变成多条流
     */
    SplitStream<SensorReading> splitStream = sensorReadingStream
        .split(new OutputSelector<SensorReading>() {
          @Override
          public Iterable<String> select(SensorReading value) {
            List<String> output = new ArrayList<String>();
            if (value.getTemperature() > 30) {
              output.add("high");
            } else {
              output.add("low");
            }
            return output;
          }
        });
    DataStream<SensorReading> highStream = splitStream.select("high");
    DataStream<SensorReading> lowStream = splitStream.select("low");
    DataStream<SensorReading> allStream = splitStream.select("low", "high");
//    lowStream.print("low");
//    highStream.print("high");

    /**
     *合并流有两种:connect,union
     * connect只能合并两条流，DataStream的类型可以不一样，输出也可以不一样
     * union而已合并多条流，但DataStream的类型要一样
     */
    DataStream<Tuple2<String, Long>> warnStream = highStream.map(
        new MapFunction<SensorReading, Tuple2<String, Long>>() {
          @Override
          public Tuple2<String, Long> map(SensorReading sensorReading) throws Exception {
            return new Tuple2<>(sensorReading.getId(), sensorReading.getTimestamp());
          }
        });
    ConnectedStreams<Tuple2<String, Long>, SensorReading> connectStream = warnStream
        .connect(lowStream);

    DataStream<Object> comapStream = connectStream
        .map(new CoMapFunction<Tuple2<String, Long>, SensorReading, Object>() {

          @Override
          public String map1(Tuple2<String, Long> value) throws Exception {
            return "高温告警[id:" + value.f0 + ",time:" + value.f1+"]";
          }

          @Override
          public SensorReading map2(SensorReading value) throws Exception {
            return value;
          }
        });
//    comapStream.print("conmap");

    DataStream<SensorReading> unionStream = highStream.union(lowStream);
    unionStream.print("union");
    env.execute("tranform test");


  }

}
